package comsumer

import (
	"errors"

	"strconv"
	"sync"
	"time"

	"fmt"
	"github.com/couchbase/gocb"
	"github.com/upmio/horus-agent/model"
	"github.com/upmio/horus-polymerizer/log"
	"github.com/upmio/horus-polymerizer/queue"
	"strings"
)

//MetricComsumerManager  is exported
type MetricComsumerManager struct {
	curnum   int
	maxnum   int
	lock     *sync.Mutex
	cburi    string
	cbpasswd string
	cbbucket string
}

//NewMetricComsumerManager  is exported
func NewMetricComsumerManager(cburi, bucketname, cbpasswd string, maxnum int) *MetricComsumerManager {
	return &MetricComsumerManager{
		curnum:   0,
		maxnum:   maxnum,
		lock:     new(sync.Mutex),
		cburi:    cburi,
		cbbucket: bucketname,
		cbpasswd: cbpasswd,
	}
}

//InitAndStartManager  is exported
func (c *MetricComsumerManager) InitAndStartManager() error {
	log.Info(" Start metric comsumers and metric manager")

	for i := 0; i < c.maxnum; i++ {
		go c.comsumerHandle(c.cburi, c.cbbucket, c.cbpasswd)
	}

	go c.manager()
	return nil
}

func getMetricKey(point *model.MetricPoint) string {
	var documentID string
	if point.History == false {
		documentID = point.Endpoint + ":" + point.Metric
	} else {
		documentID = point.Endpoint + ":" + point.Metric + ":" + strconv.Itoa(int(point.Timestamp/60))
	}
	log.Debug("the documentID", documentID)

	return documentID
}

func (c *MetricComsumerManager) addComsumer() error {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.curnum++

	if c.curnum > c.maxnum {
		return errors.New("biger than maxnum")
	}
	return nil
}

func (c *MetricComsumerManager) delComsumer() {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.curnum--
}

func (c *MetricComsumerManager) manager() {
	counter := 0
	time.Sleep(2 * 60 * time.Second)
	for {
		for i := c.curnum; i < c.maxnum; i++ {
			log.Info("restart one metric comsumer")
			go c.comsumerHandle(c.cburi, c.cbbucket, c.cbpasswd)
		}

		if counter%30 == 0 {
			log.Infof("the metricqueue infos: curlen:%d ,maxlen:%d", queue.GetMetricQcurlen(), queue.GetMetricQmaxlen())
		}

		counter++
		time.Sleep(2 * 60 * time.Second)
	}

}
func (c *MetricComsumerManager) getConnectedBucket(cburi, bucketname, passwd string) (*gocb.Bucket, error) {
	urls := strings.Split(cburi, ",")

	for _, url := range urls {
		myCluster, _ := gocb.Connect(url)
		metricBucket, err := myCluster.OpenBucket(bucketname, passwd)
		if err != nil {
			errStr := fmt.Sprintf("open the bucket fail：%s with ip(%s)", err.Error(), url)
			log.Warn(errStr)
			continue

		}
		return metricBucket, err
	}

	return nil, errors.New("try all ip but still fail")
}

func (c *MetricComsumerManager) comsumerHandle(cburi, bucketname, passwd string) {

	//operate bulket  may occur panic
	defer func() {
		if err := recover(); err != nil {
			log.Error("metric comsumer occur panic : ", err)
		}

		c.delComsumer()
	}()

	if err := c.addComsumer(); err != nil {
		log.Warn("start metric comsumer conflict:", err, " curlen:", c.curnum, "maxlen:", c.maxnum)
		return
	}

	metricBucket, err := c.getConnectedBucket(cburi, bucketname, passwd)
	if err != nil {
		log.Error("open metric bucket fail:", err)
		return
	}

	defer metricBucket.Close()

	log.Info("the Metric comsumer start")

	for {

		pointslice, _ := queue.GetMetricSlice()
		log.Debug("the metricslice len is", len(pointslice))
		var items []gocb.BulkOp
		for _, metric := range pointslice {
			if metric == nil {
				log.Debug("metric equal nil")
				continue
			}

			ekey := getMetricKey(metric)
			items = append(items, &gocb.UpsertOp{Key: ekey, Value: metric})
		}

		if err := metricBucket.Do(items); err != nil {
			log.Error("the metric bulk write to cb err:", err)

			if queue.GetMetricQcurlen()*2 < queue.GetMetricQmaxlen() {
				queue.PutMetricPoint(pointslice)
				log.Infof("reput the metricSlice to queue,the first metric :%v", pointslice[0])
			}

			break
		}
		log.Debug("wtire metric to cb ok")
	}
	log.Warn("the metirc comsumer stop")
}
